import json
import threading
import time
import queue
from multiprocessing import Process, Queue
from tools.redis_server import RedisServer
from kafka import KafkaConsumer, KafkaProducer
from tools.util import *
from process_event import process_main
from config import ConfigData, Define, KakoList
from datetime import datetime, timedelta


class Event:
    def __init__(self):
        self.config_data = ConfigData
        self.format_str = self.config_data["format_str"]
        self.gantry_info = self.config_data["gantry_info"]
        self.route_list = self.config_data["route"]
        self.save_path = self.config_data["save_path"]
        # 门架数据
        self.gantry_data = {}
        # 数据队列
        self.q = queue.Queue(maxsize=100)
        self.q_dict = {}
        self.level_dict = {}

        self.id_dict = {}
        self.info_dict = {}
        self.output = {}
        # 实时时间
        # self.current_simulated_time = None
        # 接收的最后一帧数据
        # self.last_car = None
        # redis 服务器
        # self.redis_server = RedisServer(self.config_data["redis_host"], self.config_data["redis_port"],
        #                                 self.config_data["redis_pd"], self.config_data["redis_password"])
        # kafka 服务器
        self.kafka_producer = KafkaProducer(
            bootstrap_servers=self.config_data["kafka_host"],
            key_serializer=str.encode,
            value_serializer=lambda x: json.dumps(x).encode('utf-8')
        )
        self.kafka_consumer = KafkaConsumer(
            self.config_data["kafka_cars_topic"],
            bootstrap_servers=self.config_data["kafka_host"],
            auto_offset_reset='latest',
            enable_auto_commit=False
        )
        # 线程管理
        self.thread_flag = True
        self.folder_path = None
        self.log_name = None

        self.yongdu = 0
        self.yongdu_dict = {}

        self.kako_dict = {item['pileNumber']: {key: value for key, value in item.items() if key != 'pileNumber'}
                          for item in KakoList}

    # def subscribe_data_redis(self):
    #     channel = self.config_data["redis_channel"]
    #     self.redis_server.subscribe(channel)
    #     while True:
    #         messages = self.redis_server.get_messages()
    #         if messages:
    #             try:
    #                 current_time = messages[-1][Define["TIME"]]
    #                 self.current_simulated_time = datetime.strptime(current_time, self.format_str)
    #                 # 逆序遍历messages
    #                 i = -1
    #                 while messages[i] != self.last_car and i > -len(messages):
    #                     self.q.put(messages[i], block=False)
    #                     i -= 1
    #                 self.last_car = messages[-1]
    #             except Exception as e:
    #                 print("解析数据失败！！！")
    #         time.sleep(0.1)
    #         # 退出线程
    #         if not self.thread_flag:
    #             break

    def subscribe_data_kafka(self):
        # 将消费者的偏移量设置为最后一条消息的偏移量
        while True:
            time.sleep(0.02)
            for message in self.kafka_consumer:
                if message:
                    try:
                        message_value = message.value.decode('utf-8')
                        vehicle_info = json.loads(message_value)
                        # current_time = vehicle_info[Define["TIME"]]
                        # self.current_simulated_time = datetime.strptime(current_time, self.format_str)
                        self.q.put(vehicle_info, block=False)
                    except Exception as e:
                        print("解析数据失败！！！")
            # 退出线程
            if not self.thread_flag:
                break

    def init(self):
        info_dict = {}
        self.output["data"] = {}
        self.output["gantry_id"] = {}
        T = datetime.now().strftime("%Y-%m-%d_%H-%M-%S")
        self.folder_path = os.path.join(self.save_path, T)
        if not os.path.exists(self.folder_path):
            os.makedirs(self.folder_path)
        self.log_name = "output_" + T + ".txt"
        for info in self.gantry_info.keys():
            self.q_dict[info] = Queue(maxsize=100)
            self.level_dict[info] = Queue(maxsize=100)
            self.output["data"][info] = {}
            self.yongdu_dict[info] = None
            if info[0] not in self.info_dict:
                self.info_dict[info[0]] = []
            self.info_dict[info[0]].append(info)
            info_dict[info] = [self.gantry_info[info]["up_gantry_id"], self.gantry_info[info]["down_gantry_id"]]
        # 获取门架ID对应的路段编号
        self.id_dict = format_conversion(info_dict)
        # 实时监听最新门架数据
        threading.Thread(target=self.subscribe_data_kafka).start()
        for info in self.gantry_info.keys():
            config_data = self.config_data
            config_data["gantry_info"] = self.gantry_info[info]
            config_data["gantry_info"]["info"] = info
            p = Process(target=process_main, args=(config_data, self.q_dict[info], self.level_dict[info],))
            p.start()

    def add_message(self):
        file_path = os.path.join(self.folder_path, self.log_name)
        f = open(file_path, 'w')
        while True:
            while not self.q.empty():
                message = self.q.get(timeout=0.1)
                gantry_id = message[Define["GANTRY"]]
                if gantry_id in self.id_dict:
                    info_list = self.id_dict[gantry_id]
                    for info in info_list:
                        self.q_dict[info].put(message, block=False)
                        # print(self.q_dict[info].qsize())
            for info in self.gantry_info.keys():
                gantry_up = self.gantry_info[info]['up_gantry_id']
                gantry_down = self.gantry_info[info]['down_gantry_id']
                while not self.level_dict[info].empty():
                    con = self.level_dict[info].get(timeout=0.1)
                    print(info, str(con))
                    # if info != "10003":
                    #     continue
                    tt = con['time'].timestamp()
                    if con['content'] != "畅通":
                        s = "阻断"
                        # 阻断上报
                        msg = {
                            "orgCode": gantry_down,
                            "upOrgCode": gantry_up,
                            "intervalSpeed": con['speed'],
                            "totalFlow": con['num'],
                            "congestion": s.encode('unicode-escape').decode(),
                            "globalTime": int(tt),
                            "lastTimeIndex": 1,
                            "info": "",
                            "updateTime": int(time.time()),
                            "timestamp": int(tt),
                            "direction": 1,
                            "sNum": self.kako_dict[gantry_up]['RealPileNumber'],
                            "eNum": self.kako_dict[gantry_down]['RealPileNumber'],
                            "sLon": self.kako_dict[gantry_up]['lon'],
                            "sLat": self.kako_dict[gantry_up]['lat'],
                            "eLon": self.kako_dict[gantry_down]['lon'],
                            "eLat": self.kako_dict[gantry_down]['lat'],
                            "secLevel": 1 if con['content'] == "畅通" else 5,
                            "isBreakdown": 0 if con['content'] == "畅通" else 1,
                            "eventId": 0,
                            "eventInfo": ""
                        }

                        self.kafka_producer.send(self.config_data["kafka_event_topic"], value=msg, key="key")

                    # 拥堵上报
                    if con['con'] != "畅通":
                        yongdu = {"畅通": 1, "轻度拥堵": 2, "中度拥堵": 3, "严重拥堵": 4}
                        if self.yongdu_dict[info] is None:
                            self.yongdu_dict[info] = con['time']
                            self.yongdu += 1
                        else:
                            self.yongdu_dict[info] = con['time']
                            if con['time'] - self.yongdu_dict[info] > timedelta(minutes=10):
                                self.yongdu += 1
                        msg1 = {
                            "orgCode": gantry_down,
                            "upOrgCode": gantry_up,
                            "intervalSpeed": con['speed'],
                            "totalFlow": con['num'],
                            "congestion": con['con'].encode('unicode-escape').decode(),
                            "globalTime": int(tt),
                            "lastTimeIndex": 1,
                            "info": "",
                            "updateTime": int(time.time()),
                            "timestamp": int(tt),
                            "direction": self.gantry_info[info]['direction'],
                            "sNum": self.kako_dict[gantry_up]['RealPileNumber'],
                            "eNum": self.kako_dict[gantry_down]['RealPileNumber'],
                            "sLon": self.kako_dict[gantry_up]['lon'],
                            "sLat": self.kako_dict[gantry_up]['lat'],
                            "eLon": self.kako_dict[gantry_down]['lon'],
                            "eLat": self.kako_dict[gantry_down]['lat'],
                            "secLevel": yongdu[con['con']],
                            "isBreakdown": 0,
                            "eventId": self.yongdu,
                            "eventInfo": con['con']
                        }

                        self.kafka_producer.send(self.config_data["kafka_event_topic"], value=msg1, key="key")

                    f.write(info + " " + str(con) + "\n")

            time.sleep(0.1)
